package com.zhao.streaming

import com.zhao.commonUtil.{ConnectionManager, JsonParser}
import com.zhao.util.{ParamParserUtil, SparkUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.log4j.{Level, Logger}
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.dstream.DStream

import java.sql.{Connection, Date, PreparedStatement, Timestamp}

/**
 * Description: 实时解析用户所在地<br/>
 * Copyright (c) ，2021 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2021/1/5 16:50
 *
 * @author 柒柒
 * @version : 1.0
 */

object CrowdLocationInfo {
  def main(args: Array[String]): Unit = {
    if (args.length < 3){
      System.err.println("Usage: CrowdLocationInfo <BachInterval> <windowDuration> <slideDuration>")
      System.exit(1)
    }
    Logger.getRootLogger.setLevel(Level.ERROR)

    //创建流
    val (messages, ssc) = SparkUtil.createKafkaStream(args)
//    ssc.checkpoint("/spark_checkpoint/CrowdLocationInfo/")

    //解析参数
    val paramMap: Map[String, String] = ParamParserUtil.argsToOptionMap(args)

//    messages.map(record => (record.value().toString)).print

    val crowdLocation: DStream[(String, Int)] = crowdLocationParser(messages, paramMap)

    //设置窗口长度及滑动间隔,返回窗口长度的数据流
//    val windowStream: DStream[(String, Int)] = crowdLocation.reduceByKeyAndWindow(_ + _, _ - _, Seconds(paramMap("windowDuration").toInt), Seconds(paramMap("slideDuration").toInt))

    val finalStream: DStream[(String, Int)] = crowdLocation.filter(info => {
      info != null && info._1 != null && info._2 != 0
    }).transform(rdd => {
      rdd.map(k => {
        val areaAndTown = k._1.split("_")(0)
        (areaAndTown, 1)
      })
    }).reduceByKey(_ + _)

//    finalStream.print()

    //持久化到mysql
    persistCrowLocation(finalStream)
    ssc.start()
    ssc.awaitTermination()
  }

  def persistCrowLocation(dstream: DStream[(String, Int)]) = {
    dstream.foreachRDD(rdd =>{
      rdd.foreachPartition(iter =>{
        val now: Long = System.currentTimeMillis()
        var conn: Connection = null
        var pstmt: PreparedStatement = null
        try{
          conn = ConnectionManager.getConnnection()
          conn.setAutoCommit(false)
          val sql =
            """
              |insert into fz_bigscreen_crowd_location_info (stat_date, area, town, user_count, update_time)
              |values(?,?,?,?,?) on duplicate key update stat_date=VALUES(stat_date),area=VALUES(area),
              |town=VALUES(town),user_count=VALUES(user_count) + user_count,update_time=VALUES(update_time)
              |""".stripMargin

          pstmt = conn.prepareStatement(sql)
          while (iter.hasNext){
            val data: (String, Int) = iter.next()
            val areaAndTown = data._1.split("#")
            val area = areaAndTown(0)
            val town = areaAndTown(1)
            val user_count = data._2
            pstmt.setDate(1,new Date(now))
            pstmt.setString(2,area)
            pstmt.setString(3,town)
            pstmt.setInt(4,user_count)
            pstmt.setTimestamp(5,new Timestamp(now))
            pstmt.addBatch()
          }
          pstmt.executeBatch()
          conn.commit()
        }catch {
         case e: Throwable => e.printStackTrace()
            conn.rollback()
        }finally {
          ConnectionManager.closeAll(null,pstmt,conn)
        }
      })
    })
  }

  //解析用户所处区县位置
  def crowdLocationParser(dstream: DStream[ConsumerRecord[String, String]], paramMap: Map[String, String]) = {
    val filterStream = dstream.filter(line => {
      try {
        val info: String = line.value()
        val jsonParser = new JsonParser(info)
        if (info != null && info.contains("event") && info.contains("province") && info.contains("city")
          && info.contains("area") && info.contains("distinct_id") && info.contains("toon_type") && info.contains("town")) {
          val event = jsonParser.getJsnField("event")
          val city = jsonParser.getSCJsnField("properties", "city")
          val user_id = jsonParser.getJsnField("distinct_id")
          val toon_type = jsonParser.getSCJsnField("properties", "toon_type")

          (("LocationData".equals(event) || "HFirstService".equals(event) || "H5PageView".equals(event))
            && city.contains("佛山") && user_id.trim.length <= 15
            && paramMap("toon_type").equals(toon_type))
        } else {
          false
        }
      } catch {
        case e: Throwable => e.printStackTrace()
          false
      }
    })

    val finalStream: DStream[(String, Int)] = filterStream.map(line => {
      val info: String = line.value()
      val jsonParser: JsonParser = new JsonParser(info)
      val area = jsonParser.getSCJsnField("properties", "area")
      val town = jsonParser.getSCJsnField("properties", "town")
      val user_id = jsonParser.getJsnField("distinct_id")
      (area + "#" + town + "_" + user_id, 1)
    }).transform(rdd => rdd.distinct())
    finalStream
  }
}










